package org.bigwinner.flume.sources;

import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.EventDeliveryException;
import org.apache.flume.PollableSource;
import org.apache.flume.conf.Configurable;
import org.apache.flume.event.EventBuilder;
import org.apache.flume.source.AbstractSource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.*;

/**
 * @author: IT大狮兄
 * @date: 2021/8/13 下午9:11
 * @version: 1.0.0
 * @description: 自定义Source--读取MySQL表的数据
 */
public class MysqlSource extends AbstractSource implements PollableSource, Configurable {
    private static final Logger LOGGER = LoggerFactory.getLogger(MysqlSource.class);

    private String mysqlUrl;
    private String mysqlUser;
    private String mysqlPassword;
    private String mysqlTable;
    private String mysqlDriver;
    private Connection conn = null;

    public Status process() throws EventDeliveryException {
        String sql = "select * from " + mysqlTable;
        try {
            PreparedStatement statement = conn.prepareStatement(sql);
            ResultSet resultSet = statement.executeQuery();
            while (resultSet.next()) {
                String id = resultSet.getString(1);
                String uuid = resultSet.getString(2);
                String iccid = resultSet.getString(3);
                byte[] eventBytes = new StringBuilder().append(id).append("--")
                        .append(uuid).append("--").append(iccid).toString().getBytes();
                Event event = EventBuilder.withBody(eventBytes);
                getChannelProcessor().processEvent(event);
            }
        } catch (SQLException throwables) {
            throwables.printStackTrace();
        }
        return Status.READY;
    }

    public long getBackOffSleepIncrement() {
        return 0;
    }

    public long getMaxBackOffSleepInterval() {
        return 0;
    }

    @Override
    /** Flume生命周期开始，可以做一些初始化的工作 */
    public void start() {
        LOGGER.info("Mysql source start......");
        try {
            Class.forName(mysqlDriver);
            conn = DriverManager.getConnection(mysqlUrl, mysqlUser, mysqlPassword);
        } catch (ClassNotFoundException e) {
            LOGGER.error("Driver class is not found!");
        } catch (SQLException throwables) {
            LOGGER.error("get the connection error: {}", throwables);
        }

    }

    @Override
    /** Flume生命周期结束，可以做一些保存等结束前的工作 */
    public void stop() {
        LOGGER.info("Mysql source stop......");
        if (conn != null) {
            try {
                conn.close();
            } catch (SQLException throwables) {
                LOGGER.error("连接关闭异常: {}", throwables);
            }
        }
        super.stop();
    }


    /** Flume配置文件读取的方法 */
    public void configure(Context context) {
        mysqlDriver = context.getString("mysql.driver", "");
        mysqlUrl = context.getString("mysql.url", "");
        mysqlUser = context.getString("mysql.user", "");
        mysqlPassword = context.getString("mysql.password", "");
        mysqlTable = context.getString("mysql.table", "");
        LOGGER.info("mysql_driver: {} --> mysql_url: {} --> mysql_user: {} --> mysql_password: {} --> mysql_table: {}",
                mysqlDriver, mysqlUrl, mysqlUser, mysqlPassword, mysqlTable);
    }
}
